package com.zero.stream;

import com.zero.analy.PindaoRD;
import com.zero.input.KafkaMessage;
import com.zero.stream.map.PindaoKafkaMap;
import com.zero.stream.map.PindaoRDMap;
import com.zero.stream.reduce.PindaoReduce;
import com.zero.transfer.KafkaMessageSchema;
import com.zero.transfer.KafkaMessageWatermarks;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;

public class ProcessData {


    public static void main(String[] args)throws Exception {
        args = new String[]{"--input-topic", "test","--bootstrap.servers", "118.24.113.107:9092",
                "--zookeeper.connect", "118.24.113.107:2181", "--group.id", "myconsumer1"};
        final ParameterTool parameterTool = ParameterTool.fromArgs(args);

        if (parameterTool.getNumberOfParameters() < 4) {
            System.out.println("Missing parameters!\n" +
                    "Usage: Kafka --input-topic <topic>  " +
                    "--bootstrap.servers <kafka brokers> " +
                    "--zookeeper.connect <zk quorum> --group.id <some id>");
            return;
        }

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().disableSysoutLogging();
        env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
        env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
        env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        FlinkKafkaConsumer010 flinkKafkaConsumer = new FlinkKafkaConsumer010<KafkaMessage>(
                parameterTool.getRequired("input-topic"),
                new KafkaMessageSchema(),
                parameterTool.getProperties());

        DataStream<KafkaMessage> input = env.addSource(flinkKafkaConsumer.assignTimestampsAndWatermarks(new KafkaMessageWatermarks()));
        DataStream<PindaoRD> map = input.map(new PindaoKafkaMap());
        DataStream<PindaoRD> reduce=  map.keyBy("pingdaoid").countWindow(10,5).reduce(new PindaoReduce());

        reduce.print();
        env.execute("pindaoredian");
    }
}
